// package com.jourwon.spring.boot.config;
//
// import lombok.extern.slf4j.Slf4j;
// import org.apache.kafka.clients.CommonClientConfigs;
// import org.apache.kafka.clients.admin.AdminClientConfig;
// import org.apache.kafka.clients.consumer.ConsumerConfig;
// import org.apache.kafka.common.config.SaslConfigs;
// import org.apache.kafka.common.serialization.StringDeserializer;
// import org.springframework.beans.factory.annotation.Value;
// import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
// import org.springframework.context.annotation.Bean;
// import org.springframework.context.annotation.Configuration;
// import org.springframework.kafka.annotation.EnableKafka;
// import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
// import org.springframework.kafka.core.ConsumerFactory;
// import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
// import org.springframework.kafka.listener.ContainerProperties;
//
// import java.util.HashMap;
// import java.util.Map;
//
// /**
//  * Kafka消费者配置
//  *
//  * @author JourWon
//  * @date 2022/4/21
//  */
// @Slf4j
// @EnableKafka
// @Configuration
// public class KafkaConsumerConfig /*implements DisposableBean*/ {
//
//     /**
//      * kafka服务的ip端口,集群用逗号分隔的主机:端口对列表
//      */
//     @Value("${spring.kafka.bootstrap-servers}")
//     private String bootstrapServers;
//
//     /**
//      * 以下为kafka消费者配置
//      */
//     @Value("${spring.kafka.consumer.group-id}")
//     private String groupId;
//
//     @Value("${spring.kafka.consumer.auto-offset-reset}")
//     private String autoOffsetReset;
//
//     @Value("${spring.kafka.consumer.enable-auto-commit}")
//     private Boolean enableAutoCommit;
//
//     @Value("${spring.kafka.consumer.auto-commit-interval}")
//     private Integer autoCommitInterval;
//
//     @Value("${spring.kafka.consumer.heartbeat-interval}")
//     private Integer heartbeatInterval;
//
//     @Value("${spring.kafka.consumer.fetch-min-size}")
//     private Integer fetchMinSize;
//
//     @Value("${spring.kafka.consumer.fetch-max-wait}")
//     private Integer fetchMaxWait;
//
//     @Value("${spring.kafka.consumer.max-poll-records}")
//     private Integer maxPollRecords;
//
//     @Value("${spring.kafka.consumer.properties.max.poll.interval.ms}")
//     private Integer maxPollIntervalMs;
//
//     @Value("${spring.kafka.consumer.properties.session.timeout.ms}")
//     private Integer sessionTimeoutMs;
//
//     @Value("${spring.kafka.consumer.properties.fetch.max.bytes}")
//     private Integer fetchMaxBytes;
//
//     @Value("${spring.kafka.consumer.properties.sasl.mechanism:}")
//     private String saslMechanism;
//
//     @Value("${spring.kafka.consumer.properties.security.protocol:}")
//     private String securityProtocol;
//
//     @Value("${spring.kafka.consumer.properties.sasl.jaas.config:}")
//     private String saslJaasConfig;
//
//     /**
//      * 以下为kafka监听器配置
//      */
//     @Value("${spring.kafka.listener.type}")
//     private KafkaProperties.Listener.Type type;
//
//     @Value("${spring.kafka.listener.ack-mode}")
//     private ContainerProperties.AckMode ackMode;
//
//     @Value("${spring.kafka.listener.concurrency}")
//     private Integer concurrency;
//
//     /**
//      * 以下为kafka鉴权配置
//      */
//     // @Value("${spring.kafka.jaas.login-module:}")
//     // private String jaasLoginModule;
//
//     @Value("${spring.kafka.jaas.enabled:false}")
//     private Boolean enabled;
//
//     // @Value("${spring.kafka.jaas.options.username:}")
//     // private String jaasUsername;
//     //
//     // @Value("${spring.kafka.jaas.options.password:}")
//     // private String jaasPassword;
//
//     // @PostConstruct
//     // public void initJassFile() throws IOException {
//     //     if (Boolean.FALSE.equals(enabled)) {
//     //         return;
//     //     }
//     //
//     //     File file = FileUtils.getFile(defaultJaasFileName());
//     //     file.deleteOnExit();
//     //
//     //     String jaasFileContent = "KafkaClient {\n" +
//     //             "\t" + jaasLoginModule + " required\n" +
//     //             "\tusername=\"" + jaasUsername + "\"\n" +
//     //             "\tpassword=\"" + jaasPassword + "\";\n" +
//     //             "};";
//     //
//     //     if (!file.exists()) {
//     //         FileUtils.writeStringToFile(file, jaasFileContent, StandardCharsets.UTF_8);
//     //     }
//     //
//     //     System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, file.getAbsolutePath());
//     //
//     //     log.info("KAFKA JASS FILE地址:{},配置:{}", file.getAbsolutePath(), jaasFileContent);
//     // }
//     //
//     // private String defaultJaasFileName() {
//     //     return System.getProperty("user.dir") + File.separator + "kafka_client_jaas.conf";
//     // }
//     //
//     // @Override
//     // public void destroy() {
//     //     if (Boolean.FALSE.equals(enabled)) {
//     //         return;
//     //     }
//     //     System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
//     // }
//
//     @Bean
//     public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory consumerFactory) {
//         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
//
//         factory.setConsumerFactory(consumerFactory);
//         // 消费者组中线程数量
//         factory.setConcurrency(concurrency);
//         // 当使用批量监听器时需要设置为true
//         if (type.equals(KafkaProperties.Listener.Type.BATCH)) {
//             factory.setBatchListener(true);
//         }
//         factory.getContainerProperties().setAckMode(ackMode);
//         factory.getContainerProperties().setPollTimeout(maxPollIntervalMs);
//
//         return factory;
//     }
//
//     @Bean
//     public ConsumerFactory<String, String> consumerFactory() {
//         return new DefaultKafkaConsumerFactory<>(consumerConfigs());
//     }
//
//     private Map<String, Object> consumerConfigs() {
//         Map<String, Object> propsMap = new HashMap<>(16);
//         propsMap.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//
//         propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
//         propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
//         propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
//         propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
//         propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
//         propsMap.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, heartbeatInterval);
//         propsMap.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, fetchMaxWait);
//         propsMap.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, fetchMinSize);
//         propsMap.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchMaxBytes);
//         propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs);
//         propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
//         propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//
//         if (Boolean.TRUE.equals(enabled)) {
//             propsMap.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
//             propsMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
//             propsMap.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);
//         }
//
//         return propsMap;
//     }
//
// }
